import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector

object StateProgramingTest02 {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream: DataStream[String] = env.socketTextStream("localhost", 7777)

    val tempDS = stream.map(
      data => {
        val dataArr = data.split(",")
        TemperatureData(dataArr(0), dataArr(1).toLong, dataArr(2).toDouble)
      }
    )

    val keyedData = tempDS.keyBy(_.id)

    //TODO flatMapWithState
    keyedData
      .flatMapWithState[(String, Double, Double), Double] {
        case (in: TemperatureData, None) =>
          (List.empty, Some(in.temp))
        case (r: TemperatureData, lastTemp: Some[Double]) =>
          val tempDiff = (r.temp - lastTemp.get).abs
          if (tempDiff > 1.7) {
            (List((r.id, r.temp, tempDiff)), Some(r.temp))
          } else {
            (List.empty, Some(r.temp))
          }
      }

  }
  case class TemperatureData(id: String, time: Long, temp: Double)
}
